#pragma once

#include "muduo/proto/dispatcher.h"
#include "muduo/proto/codec.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpClient.h"
#include "muduo/base/CountDownLatch.h"

#include "../common/helper.hpp"
#include "../common/log.hpp"
#include "../common/mq_proto/mq_proto.pb.h"
#include "../common/mq_msg/mq_msg.pb.h"
#include "../common/consumer.hpp"
#include "mq_channel.hpp"
#include "mq_worker.hpp"

#include <iostream>
#include <functional> // bind()
#include <memory>

namespace xzt
{
    using std::string;
    using namespace std::placeholders;

    class Connection
    {
    public:
        using MessagePtr = std::shared_ptr<google::protobuf::Message>;
        using ptr = std::shared_ptr<Connection>;

        Connection(const string &sip, int port, const AsyncWorker::ptr &worker) : 
                                              _latch(1), _client(
                                                             worker->_loopthread.startLoop(),
                                                             muduo::net::InetAddress(sip, port),
                                                             "Connection"),
                                              _dispatcher(
                                                  std::bind(&Connection::onUnKnownMessage,
                                                            this,
                                                            _1, _2, _3)),
                                              _codec(std::make_shared<ProtobufCodec>(
                                                  std::bind(&ProtobufDispatcher::onProtobufMessage,
                                                            &_dispatcher,
                                                            _1, _2, _3))),
                                              _channel_manager(std::make_shared<ChannelManager>()),
                                              _worker(worker)
        {
            // 注册业务处理函数
            _dispatcher.registerMessageCallback<xzt::basicCommonReponse>(std::bind(
                &Connection::basicResponse, this,
                _1, _2, _3));
            _dispatcher.registerMessageCallback<xzt::basicConsumeResponse>(std::bind(
                &Connection::consumeResponse, this,
                _1, _2, _3));

            // 将类成员函数设置为服务器的回调处理函数
            _client.setConnectionCallback(std::bind(
                &Connection::onConnection,
                this, _1));

            _client.setMessageCallback(std::bind(
                &ProtobufCodec::onMessage,
                _codec, _1, _2, _3));

            // 连接服务器
            connect();
        }

        Channel::ptr openChannel()
        {
            // 1. 创建信道
            Channel::ptr channel = _channel_manager->create(_conn, _codec);
            // 2. 向服务器发送打开信道的请求
            bool ret = channel->openChannel();
            if (!ret)
            {
                ELOG("打开信道失败");
                return nullptr;
            }
            return channel;
        }

        void closeChannel(const Channel::ptr &channel)
        {
            channel->closeChannel();
            DLOG("关闭信道: %s", channel->cid().c_str());
            _channel_manager->remove(channel->cid());
        }

    private:
        void connect()
        {
            _client.connect();
            _latch.wait();
        }

        // 回调函数
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            // 新连接建立成功时的回调函数
            if (conn->connected())
            {
                _latch.countDown();
                _conn = conn;
            }
            else
            {
                _conn.reset();
            }
        }

        void onUnKnownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            ILOG("onUnKnownMessage: %s", message->GetTypeName().c_str());
        }

        void basicResponse(const muduo::net::TcpConnectionPtr &conn, const basicCommonReponsePtr &message, muduo::Timestamp)
        {
            // 1. 找到对应的信道
            Channel::ptr channel = _channel_manager->get(message->cid());
            if (channel == nullptr)
            {
                ELOG("未找到信道");
                return;
            }
            // 2. 将得到的相应，放入信道对应的基础相应hash_map中
            channel->putOnBasicResp(message);
        }

        void consumeResponse(const muduo::net::TcpConnectionPtr &conn, const basicConsumeResponsePtr &message, muduo::Timestamp)
        {
            // 1. 找到对应的信道
            Channel::ptr channel = _channel_manager->get(message->cid());
            if (channel == nullptr)
            {
                ELOG("未找到信道");
                return;
            }
            // 2. 封装异步任务（消息处理任务），抛入异步工作线程池
            _worker->_threadpool.push(std::bind(&Channel::consume, channel, message));  // 这里channel是this指针, message是参数
        }

        // 客户端->服务器 发送信息，类型为google::protobuf::Message
        bool send(const google::protobuf::Message *pmsg)
        {
            if (_conn->connected())
            {
                _codec->send(_conn, *pmsg);
                return true;
            }
            return false;
        }

    private:
        // 为connect提供异步阻塞，用于实现同步
        muduo::CountDownLatch _latch;
        // _conn是客户端对应的连接
        muduo::net::TcpConnectionPtr _conn;
        // _client是客户端对象
        muduo::net::TcpClient _client;
        // 请求分发器
        ProtobufDispatcher _dispatcher;
        // 协议处理器
        ProtobufCodecPtr _codec;
        // 信道管理器
        ChannelManager::ptr _channel_manager;
        // 异步工作
        AsyncWorker::ptr _worker;
    };
}